package com.jfree.zuul;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        demo3();
    }

public static void test1() {
    Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            System.out.println("=========================currentThread name: " + Thread.currentThread().getName());
            e.onNext(1);
            e.onComplete();
            e.onNext(2);
            e.onComplete();
            e.onNext(3);
            e.onComplete();
        }
    });
    Observer observer = new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("======================onSubscribe");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("======================onNext " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("======================onError");
        }

        @Override
        public void onComplete() {
            System.out.println("======================onComplete");
        }
    };
    observable.subscribe(observer);
}

public static void  demo2(){
    Observable
            .create(new ObservableOnSubscribe<Integer>() {
                @Override
                public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                    int i = 0;
                    while (i<100) {
                        i++;
                        e.onNext(i);
                    }
                }
            })
            .subscribeOn(Schedulers.newThread())
            .observeOn(Schedulers.newThread())
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(integer);
                }
            });
}
    public static void demo3() throws InterruptedException {
        Flowable flowable=     Flowable.create(new FlowableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                        System.out.println("发射----> 1"+"threadName:"+Thread.currentThread().getName());
                        e.onNext(1);
                        e.onNext(2);
                        e.onNext(3);
                        e.onNext(4);
                       e.onComplete();


                    }
                }, BackpressureStrategy.BUFFER) ;//create方法中多了一个BackpressureStrategy类型的参数
        Subscriber subscriber=   new Subscriber<Integer>() {


            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("接收----> " + integer+"threadName:"+Thread.currentThread().getName());
            }

            @Override
            public void onError(Throwable t) {
                System.out.println("----------------err--------------");
            }

            @Override
            public void onComplete() {
                System.out.println("接收----> 完成threadName:"+Thread.currentThread().getName());
            }
        };
        Executor er=flowableExecutor();

 for (int i=0;i<100;i++){
     flowable  .subscribeOn(Schedulers.from(er))//为上下游分别指定各自的线程
             .observeOn(Schedulers.from(er))
             .subscribe(subscriber);
        }
        Thread.sleep(50000);
        System.out.println("a=="+Df.a);
    }

    public static Executor flowableExecutor(){ //TODO 后续core，max ，alivetime，capacity要可配置不能直接写死
        return new ThreadPoolExecutor(1,1,5, TimeUnit.MINUTES,
                new ArrayBlockingQueue<Runnable>(1), Executors.defaultThreadFactory(),new RejectedExecutionHandler(){
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.out.println("--------ett flowJFerr-----------");
            }
        });
    }
    public static void demo4() {
        Flowable
                .create((FlowableOnSubscribe<Integer>) e -> {
                    for (int i = 1; i <= 100; i++) {
                        e.onNext(i);
                    }
                    e.onComplete();
                }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(Long.MAX_VALUE);            //注意此处，暂时先这么设置
                    }

                    @Override
                    public void onNext(Integer integer) {

                        System.out.println(integer+Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("接收----> 完成");
                    }
                });
    }

}
